package com.cjfather.redis.steam.listener;
import java.time.Duration;
import java.util.Vector;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.context.ApplicationContext;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.stream.Consumer;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.ObjectRecord;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;
import org.springframework.data.redis.stream.StreamMessageListenerContainer.StreamMessageListenerContainerOptions;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;

import com.fasterxml.jackson.annotation.JsonAutoDetect.Visibility;
import com.cjfather.redis.steam.annotation.constants.RedisConstants;
import com.cjfather.redis.steam.annotation.serializer.FurySerializer;
import com.cjfather.redis.steam.consumers.ConsumerI;
import com.cjfather.redis.steam.controller.PropellingInfo;
import com.cjfather.redis.steam.handler.CustomErrorHandler;
import com.fasterxml.jackson.annotation.PropertyAccessor;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.ObjectMapper.DefaultTyping;

import cn.hutool.json.JSONArray;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import lombok.extern.slf4j.Slf4j;

/**
 * 
 * @author: todd
 * @date:   2023-08-08
 */


@Slf4j
@Component
public class StreamConsumerListener implements ApplicationRunner, DisposableBean {

    @Autowired
    RedisConnectionFactory redisConnectionFactory;

    @Autowired
    ApplicationContext applicationContext;

    @Autowired
    ThreadPoolTaskExecutor threadPoolTaskExecutor;

    @Autowired
    StringRedisTemplate stringRedisTemplate;
    
    @Value("${cjfather.mq}")
    String mq;

    private Vector<StreamMessageListenerContainer<String, MapRecord<String, Object, Object>>> listenerContainers = new Vector<>();
    
    @Override
    public void run(ApplicationArguments args) {
    	JSONArray mqs = JSONUtil.parseArray(mq);
    	
    	for(JSONObject keys:mqs.toList(JSONObject.class)) {
    		String key = keys.getStr("key");
    		StreamMessageListenerContainer<String, MapRecord<String, Object, Object>> streamMessageListenerContainer = this.getContainer(key);
    		
    		for(JSONObject group:keys.getJSONArray("groups").toList(JSONObject.class)) {
    	    	String gname = group.getStr("group");	
    	    	
    	    	for(String consumer:group.getJSONArray("consumers").toList(String.class)) {
    	    		Consumer consumerOne = Consumer.from(gname, consumer);
                    StreamOffset<String> stringStreamOffset = StreamOffset.create(key, ReadOffset.lastConsumed());
                    
                    StreamListener<String, MapRecord<String, Object, Object>> streamListener = getType(key,gname,consumer);
                    
                    StreamMessageListenerContainer.ConsumerStreamReadRequest<String> streamReadRequest = StreamMessageListenerContainer
                            .StreamReadRequest
                            .builder(stringStreamOffset)
                            .consumer(consumerOne)
                            // 获取消息的过程或获取到消息给具体的消息者处理的过程中，发生了异常的处理
                            .errorHandler(new CustomErrorHandler(stringRedisTemplate, key, gname,streamListener.getClass().getName()))
                            .autoAcknowledge(true)
                            // 如果消费者发生了异常，判断是否取消消费者消费
                            .cancelOnError(throwable -> false)
                            .build();
                    streamMessageListenerContainer.register(streamReadRequest, streamListener);
                    listenerContainers.add(streamMessageListenerContainer);
                    log.info("队列{} 组{} 消费者{}",key,gname,consumer);
    	    	}
                
                try {
    				stringRedisTemplate.opsForStream().createGroup(key,ReadOffset.from("0-0"), gname);
    			} catch (Exception e) {
    				log.info("{}组存在",gname);
    			}
    		}
    		streamMessageListenerContainer.start();
    	}
    	
        log.info("有{}监听已启动",listenerContainers.size());
    }

    /**
     *  通过对应的组名来获取对应消息处理对象
     * @param type 组名
     * @return 消息处理对象
     */
    public StreamListener<String, MapRecord<String, Object, Object>> getType(String type,String group,String consumer){
//    	group：	group":"cjfather:group:1.UserStreamConsumer，实现类在com.cjfather.redis.steam.consumers包下
    	String[] gclasss = group.split("\\.");
    	if(gclasss == null || gclasss.length != 2) {
    		log.error("配置文件group名不正确");
    	}else {
    		
    		ConsumerI bean = (ConsumerI) applicationContext.getBean(gclasss[1]);
        	bean.setConsumerName(consumer);
        	bean.setGroup(group);
        	return bean;
        	
//        	return UserStreamConsumer.builder().group(group).consumerName(consumer).redisTemplate(stringRedisTemplate).build();
    	}
        if ("cjfather:queue:1".equals(type)){
        	
        }
       
        return null;
    }

    /**
     *  创建StreamMessageListenerContainerOptions对象
     */
    private StreamMessageListenerContainerOptions<String, MapRecord<String, String, String>> options(String group,String consumer){
//        // 设置线程的前缀
        threadPoolTaskExecutor.setThreadNamePrefix(group+"-");
        // 设置线程执行的线程名
        threadPoolTaskExecutor.setThreadGroupName(consumer);
        return  StreamMessageListenerContainerOptions
                .builder()
                // 一次性最多拉取多少条消息
                .batchSize(1000)
                // 执行消息轮询的执行器
                .executor(this.threadPoolTaskExecutor)
                // 超时时间，设置为0，表示不超时（超时后会抛出异常）
                .pollTimeout(Duration.ZERO)
                // 序列化器
                .serializer(new StringRedisSerializer())
                .build();
    }
    
    public StreamMessageListenerContainer<String, MapRecord<String, Object, Object>> getContainer(String key) {
    	AtomicInteger index = new AtomicInteger(1);
        int processors = Runtime.getRuntime().availableProcessors() * 5;
        ThreadPoolExecutor executor = new ThreadPoolExecutor(processors, processors, 0, TimeUnit.SECONDS,
                new LinkedBlockingDeque<>(), r -> {
            Thread thread = new Thread(r);
            thread.setName(key + "-" + index.getAndIncrement());
            thread.setDaemon(true);
            return thread;
        });
        
        Jackson2JsonRedisSerializer<Object> jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer<Object>(Object.class);
		ObjectMapper om = new ObjectMapper();
		om.setVisibility(PropertyAccessor.ALL, Visibility.ANY);
		om.enableDefaultTyping(DefaultTyping.NON_FINAL);
//		om.activateDefaultTyping(LaissezFaireSubTypeValidator.instance ,
//                ObjectMapper.DefaultTyping.NON_FINAL, JsonTypeInfo.As.PROPERTY);
		jackson2JsonRedisSerializer.setObjectMapper(om);
		
        StreamMessageListenerContainerOptions<String, MapRecord<String, Object, Object>> options =
                StreamMessageListenerContainer.StreamMessageListenerContainerOptions
                .builder()
                // 一次最多获取多少条消息
                .batchSize(RedisConstants.POLL_BATCH_SIZE)
                // 运行 Stream 的 poll task
                .executor(executor)
                // Stream 中没有消息时，阻塞多长时间，需要比 `spring.redis.timeout` 的时间小
                .pollTimeout(Duration.ofSeconds(1))
                .serializer(new StringRedisSerializer()).hashValueSerializer(jackson2JsonRedisSerializer)
                .build();
        
        return StreamMessageListenerContainer.create(this.redisConnectionFactory, options);
    }
    
    public StreamMessageListenerContainer<String, ObjectRecord<String, String>> getContainer2(String key) {
    	AtomicInteger index = new AtomicInteger(1);
        int processors = Runtime.getRuntime().availableProcessors() * 5;
        ThreadPoolExecutor executor = new ThreadPoolExecutor(processors, processors, 0, TimeUnit.SECONDS,
                new LinkedBlockingDeque<>(), r -> {
            Thread thread = new Thread(r);
            thread.setName(key + "-" + index.getAndIncrement());
            thread.setDaemon(true);
            return thread;
        });
        
        
        StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, ObjectRecord<String, String>> options =
                StreamMessageListenerContainer.StreamMessageListenerContainerOptions
                .builder()
                // 一次最多获取多少条消息
                .batchSize(RedisConstants.POLL_BATCH_SIZE)
                // 运行 Stream 的 poll task
                .executor(executor)
                // Stream 中没有消息时，阻塞多长时间，需要比 `spring.redis.timeout` 的时间小
                .pollTimeout(Duration.ofSeconds(1))
                .serializer(new StringRedisSerializer())
                .targetType(String.class)
                .build();
        
        return StreamMessageListenerContainer.create(this.redisConnectionFactory, options);
    }
    
    public StreamMessageListenerContainer<String, MapRecord<String, String, String>> getContainer3(String key) {
    	AtomicInteger index = new AtomicInteger(1);
        int processors = Runtime.getRuntime().availableProcessors() * 5;
        ThreadPoolExecutor executor = new ThreadPoolExecutor(processors, processors, 0, TimeUnit.SECONDS,
                new LinkedBlockingDeque<>(), r -> {
            Thread thread = new Thread(r);
            thread.setName(key + "-" + index.getAndIncrement());
            thread.setDaemon(true);
            return thread;
        });
        
        StreamMessageListenerContainerOptions<String, MapRecord<String, String, String>> options =
                StreamMessageListenerContainer.StreamMessageListenerContainerOptions
                .builder()
                // 一次最多获取多少条消息
                .batchSize(RedisConstants.POLL_BATCH_SIZE)
                // 运行 Stream 的 poll task
                .executor(executor)
                // Stream 中没有消息时，阻塞多长时间，需要比 `spring.redis.timeout` 的时间小
                .pollTimeout(Duration.ofSeconds(1))
                .serializer(new StringRedisSerializer())
                .build();
        
        return StreamMessageListenerContainer.create(this.redisConnectionFactory, options);
    }
    
    public StreamMessageListenerContainer<String, MapRecord<String, Object, Object>> getContainer4(String key) {
    	AtomicInteger index = new AtomicInteger(1);
        int processors = Runtime.getRuntime().availableProcessors() * 5;
        ThreadPoolExecutor executor = new ThreadPoolExecutor(processors, processors, 0, TimeUnit.SECONDS,
                new LinkedBlockingDeque<>(), r -> {
            Thread thread = new Thread(r);
            thread.setName(key + "-" + index.getAndIncrement());
            thread.setDaemon(true);
            return thread;
        });
        
        FurySerializer serializer = new FurySerializer();
		
        StreamMessageListenerContainerOptions<String, MapRecord<String, Object, Object>> options =
                StreamMessageListenerContainer.StreamMessageListenerContainerOptions
                .builder()
                // 一次最多获取多少条消息
                .batchSize(RedisConstants.POLL_BATCH_SIZE)
                // 运行 Stream 的 poll task
                .executor(executor)
                // Stream 中没有消息时，阻塞多长时间，需要比 `spring.redis.timeout` 的时间小
                .pollTimeout(Duration.ofSeconds(1))
                .keySerializer(RedisSerializer.string())
                .hashKeySerializer(RedisSerializer.string())
                .hashValueSerializer(RedisSerializer.json())
//                .targetType(PropellingInfo.class)
                .build();
        return StreamMessageListenerContainer.create(this.redisConnectionFactory, options);
    }
    
    public StreamMessageListenerContainer<String, ObjectRecord<String, PropellingInfo>> getContainer5(String key) {
    	AtomicInteger index = new AtomicInteger(1);
        int processors = Runtime.getRuntime().availableProcessors() * 5;
        ThreadPoolExecutor executor = new ThreadPoolExecutor(processors, processors, 0, TimeUnit.SECONDS,
                new LinkedBlockingDeque<>(), r -> {
            Thread thread = new Thread(r);
            thread.setName(key + "-" + index.getAndIncrement());
            thread.setDaemon(true);
            return thread;
        });
        
        StreamMessageListenerContainerOptions<String, ObjectRecord<String, PropellingInfo>> options =
                StreamMessageListenerContainer.StreamMessageListenerContainerOptions
                .builder()
                // 一次最多获取多少条消息
                .batchSize(RedisConstants.POLL_BATCH_SIZE)
                // 运行 Stream 的 poll task
                .executor(executor)
                // Stream 中没有消息时，阻塞多长时间，需要比 `spring.redis.timeout` 的时间小
                .pollTimeout(Duration.ofSeconds(1))
                .targetType(PropellingInfo.class)
                .build();
        return StreamMessageListenerContainer.create(this.redisConnectionFactory, options);
    }
    

    @Override
    public void destroy(){
        log.info("程序：destroy、关闭{}个监听器",listenerContainers.size());
        this.listenerContainers.forEach(StreamMessageListenerContainer::stop);
    }
}
